package com.icebartech.core.components;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.ScanOptions.ScanOptionsBuilder;
import org.springframework.stereotype.Component;

import com.icebartech.core.conf.SiteProperties;
import com.icebartech.core.utils.DateTimeUtility;


/**
 * redis操作组件
 * @author wenhsh
 *
 */
@Component
public class RedisComponent {
    
    private static Logger logger = LoggerFactory.getLogger("redisComponent");
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private SiteProperties siteProperties;
    
    private final static String GROUP_SEPERATOR = ":";
    
    
    /**
     * 设置不过期的缓存到redis
     * @param group 缓存分组名
     * @param key 缓存key值
     * @param value 缓存value值
     */
    public void putRedisCache(final String group, final String key, final String value){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            redisTemplate.execute(new RedisCallback<Object>(){
                @Override
                public Object doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    connection.set(redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key), redisTemplate.getStringSerializer().serialize(value));
                    return true;
                }
            });
        }
    }
    
    /**
     * 设置不过期的缓存到list，从队列后端插入
     * @param group
     * @param key
     * @param value
     */
    public void putRedisListCache(final String group, final String key, final String value){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            redisTemplate.execute(new RedisCallback<Object>(){
                @Override
                public Object doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    connection.rPush(keyBytes, valueBytes);
                    return true;
                }
                
            });
        }
    }
    
    
    /**
     * 设置不过期的缓存到set
     * @param group
     * @param key
     * @param value
     */
    public void putRedisSetCache(final String group, final String key, final String value){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            redisTemplate.execute(new RedisCallback<Object>(){
                @Override
                public Object doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    connection.sAdd(keyBytes, valueBytes);
                    return true;
                }
            });
        }
    }
    
    /**
     * 设置不过期的缓存到list，从队列前端插入
     * @param group
     * @param key
     * @param value
     */
    public void putRedisListLeftCache(final String group, final String key, final String value){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            redisTemplate.execute(new RedisCallback<Object>(){
                @Override
                public Object doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    connection.lPush(keyBytes, valueBytes);
                    return true;
                }
                
            });
        }
    }
    
    /**
     * 设置不过期的缓存到list，从队列右侧插入
     * @param group
     * @param key
     * @param value
     */
    public void putRedisListRightCache(final String group, final String key, final String value){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            redisTemplate.execute(new RedisCallback<Object>(){
                @Override
                public Object doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    connection.rPush(keyBytes, valueBytes);
                    return true;
                }
                
            });
        }
    }
    
    /**
     * 获取某key对应的list缓存下缓存数量
     * @param group
     * @param key
     * @return
     */
    public Long getRedisListCount(final String group, final String key){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            return redisTemplate.execute(new RedisCallback<Long>(){
                @Override
                public Long doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    return connection.lLen(keyBytes);
                }});
        }else{
            return 0L;
        }
    }
    
    /**
     * 获取某key对应的set缓存下缓存数量
     * @param group
     * @param key
     * @return
     */
    public Long getRedisSetCount(final String group, final String key){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            return redisTemplate.execute(new RedisCallback<Long>(){
                @Override
                public Long doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    return connection.sCard(keyBytes);
                }});
        }else{
            return 0L;
        }
    }
    
    /**
     * 获取set中所有的缓存
     * @param group
     * @param key
     * @return
     */
    public List<String> getRedisSetList(final String group, final String key){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            return redisTemplate.execute(new RedisCallback<List<String>>(){
                @Override
                public List<String> doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    Set<byte[]> byteList = connection.sMembers(keyBytes);
                    if(CollectionUtils.isNotEmpty(byteList)){
                        List<String> rtnList = new ArrayList<String>();
                        for(byte[] b : byteList){
                            String value = redisTemplate.getStringSerializer().deserialize(b);
                            rtnList.add(value);
                        }
                        return rtnList;
                    }
                    return null;
                }
            });
        }else{
            return null;
        }
    }
    
    /**
     * 从list队列的左侧出队一个元素
     * @param group 分组名称
     * @param key 缓存key
     * @return
     */
    public String getRedisListLeftPop(final String group, final String key){
    	if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) ){
            return redisTemplate.execute(new RedisCallback<String>(){
                @Override
                public String doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = connection.lPop(keyBytes);
                    if(valueBytes != null && valueBytes.length > 0){
                    	String value = redisTemplate.getStringSerializer().deserialize(valueBytes);
                    	return value;
                    }
                    return null;
                }
            });
        }
    	return null;
    }
    
    /**
     * 分页获取list中的缓存列表
     * @param group
     * @param key
     * @param pageIndex
     * @param pageSize
     * @return
     */
    public List<String> getRedisListPage(final String group, final String key, long pageIndex, long pageSize){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && pageSize > 0){
            if(pageIndex <= 0){
                pageIndex = 1;
            }
            final long start = (pageIndex - 1 ) * pageSize;
            final long end = start + (pageSize - 1);
            return redisTemplate.execute(new RedisCallback<List<String>>(){
                
                @Override
                public List<String> doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    List<byte[]> byteList = connection.lRange(keyBytes, start, end);
                    if(CollectionUtils.isNotEmpty(byteList)){
                        List<String> rtnList = new ArrayList<String>();
                        for(byte[] b : byteList){
                            String value = redisTemplate.getStringSerializer().deserialize(b);
                            rtnList.add(value);
                        }
                        return rtnList;
                    }
                    return null;
                }});
        }
        return null;
    }
    
    /**
     * 根据index索引获取list缓存中对应的value
     * @param group
     * @param key
     * @param index
     * @return
     */
    public String getRedisListIndex(final String group, final String key, final long index){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            return redisTemplate.execute(new RedisCallback<String>(){
                @Override
                public String doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = connection.lIndex(keyBytes, index);
                    String value = redisTemplate.getStringSerializer().deserialize(valueBytes);
                    return value;
                }
                
            });
        }else{
            return null;
        }
    }
    
    /**
     * 更新list缓存指定index位置的value
     * @param group
     * @param key
     * @param index
     * @param value
     */
    public void updateRedisListValue(final String group, final String key, final long index, final String value){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            redisTemplate.execute(new RedisCallback<Object>(){
                @Override
                public Object doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    connection.lSet(keyBytes, index, valueBytes);
                    return true;
                }
            });
        }
    }
    
    /**
     * 移除list中的index索引位置的缓存记录
     * @param group
     * @param key
     * @param index
     */
    public void deleteRedisListValue(final String group, final String key, final long index){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            redisTemplate.execute(new RedisCallback<Long>(){
                @Override
                public Long doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = connection.lIndex(keyBytes, index);
                    if(valueBytes != null && valueBytes.length > 0){
                        return connection.lRem(keyBytes, 0, valueBytes);
                    }else{
                        return 0L;
                    }
                }
            });
        }
    }
    
    /**
     * 迁移set中指定value的缓存记录
     * @param group
     * @param key
     * @param value
     */
    public void deleteRedisSetValue(final String group, final String key, final String value){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            redisTemplate.execute(new RedisCallback<Long>() {
                @Override
                public Long doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    return connection.sRem(keyBytes, valueBytes);
                }
            });
        }
    }
    
    /**
     * 设置一定时间内过期的缓存到redis
     * @param group 分组名
     * @param key 缓存key值
     * @param value 缓存value
     * @param expires 过期时间，以秒为单位
     */
    public void putRedisCache(final String group, final String key, final String value, final long expires){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            if(expires <= 0){
                putRedisCache(group, key, value);
            }else{
                redisTemplate.execute(new RedisCallback<Object>(){
                    @Override
                    public Object doInRedis(RedisConnection connection) throws DataAccessException{
                        connection.set(redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key), redisTemplate.getStringSerializer().serialize(value));
                        connection.expire(redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key), expires);
                        return true;
                    }
                });
            }
        }
    }
    
    /**
     * 根据key值从分组中获取缓存value
     * @param group 分组名
     * @param key 缓存key
     * @return 缓存value
     */
    public String getRedisCache(final String group, final String key){
        String rtnValue = null;
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            rtnValue = redisTemplate.execute(new RedisCallback<String>(){
                @Override
                public String doInRedis(RedisConnection connection) throws DataAccessException{
                    byte[] rtnKey = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    if(connection.exists(rtnKey)){
                        byte[] valueByte = connection.get(rtnKey);
                        String value = redisTemplate.getStringSerializer().deserialize(valueByte);
                        return value;
                    }else{
                        return null;
                    }
                }
            });
        }
        return rtnValue;
    }
    
    
    /**
     * 根据缓存key值删除该分组下的value值
     * @param group 分组名
     * @param key 缓存key值
     */
    public void deleteRedisCache(final String group, final String key){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            redisTemplate.execute(new RedisCallback<Object>(){
                @Override
                public Object doInRedis(RedisConnection connection){
                    connection.del(redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key));
                    return true;
                }
            });
        }
    }
    
    /**
     * 原子计数器+1
     * @param group 分组名
     * @param key 计数器key值
     */
    public Long incRedisKey(final String group, final String key){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            return redisTemplate.execute(new RedisCallback<Long>(){
                @Override
                public Long doInRedis(RedisConnection connection){
                    return connection.incr(redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key));
                }
            });
        }else{
            return null;
        }
    }
    
    /**
     * 原子计数器+n
     * @param group 分组名
     * @param key 计数器key值
     * @param count 计数器增加的数量
     */
    public Long incRedisKey(final String group, final String key, final long count){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && count > 0){
            return redisTemplate.execute(new RedisCallback<Long>(){
                @Override
                public Long doInRedis(RedisConnection connection){
                    return connection.incrBy(redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key), count);
                }
            });
        }else{
            return null;
        }
    }
    
    /**
     * 原子计数器-1
     * @param group 分组名
     * @param key 计数器key值
     */
    public Long decRedisKey(final String group, final String key){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            return redisTemplate.execute(new RedisCallback<Long>(){
                @Override
                public Long doInRedis(RedisConnection connection){
                    return connection.decr(redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key));
                }
            });
        }else{
            return null;
        }
    }
    
    
    /**
     * 原子计数器-n
     * @param group
     * @param key
     */
    public Long decRedisKey(final String group, final String key, final long count){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && count > 0){
            return redisTemplate.execute(new RedisCallback<Long>(){
                @Override
                public Long doInRedis(RedisConnection connection){
                    return connection.decrBy(redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key), count);
                }
            });
        }else{
            return null;
        }
    }
    
    /**
     * 单独设置某一个key的过期时间，时间单位为秒
     * @param key
     * @param seconds -1的时候即把key的有效期变成永久
     */
    public boolean setKeyExpire(final String group, final String key, final long seconds){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && seconds >= -1){
            return redisTemplate.execute(new RedisCallback<Boolean>() {
                @Override
                public Boolean doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    return connection.expire(keyBytes, seconds);
                }
                
            });
        }else{
            return false;
        }
    }
    
    /**
     * 检查一个key是否存在
     * @param group
     * @param key
     * @return
     */
    public boolean isKeyExists(final String group, final String key){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            return redisTemplate.execute(new RedisCallback<Boolean>() {
                @Override
                public Boolean doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    return connection.exists(keyBytes);
                }
                
            });
        }else{
            return false;
        }
    }
    
    /**
     * 检查对应的value在key set中是否存在
     * @param group
     * @param key
     * @param value
     * @return
     */
    public boolean isSetValueExists(final String group, final String key, final String value){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            return redisTemplate.execute(new RedisCallback<Boolean>() {
                @Override
                public Boolean doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    return connection.sIsMember(keyBytes, valueBytes);
                }
            });
        }else{
            return false;
        }
    }
    
    /**
     * 将一个key-value对添加到redis hash缓存中
     * @param group 分组名
     * @param cacheKey 缓存key
     * @param hashKey key-value对的key
     * @param hashValue key-value对的value
     * @return
     */
    public boolean putRedisHashCache(final String group, final String cacheKey, final String hashKey, final byte[] hashValue){
        boolean rtnFlag =false;
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(cacheKey) && StringUtils.isNotBlank(hashKey) && hashValue != null && hashValue.length > 0){
            rtnFlag = redisTemplate.execute(new RedisCallback<Boolean>() {
                @Override
                public Boolean doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + cacheKey);
                    Map<byte[], byte[]> hashMap = new HashMap<byte[], byte[]>();
                    byte[] hashKeyBytes = redisTemplate.getStringSerializer().serialize(hashKey);
                    hashMap.put(hashKeyBytes, hashValue);
                    connection.hMSet(keyBytes, hashMap);
                    return true;
                }
            });
        }
        return rtnFlag;
    }
    
    /**
     * 根据key获取对应的hash缓存中的value值
     * @param group 分组名
     * @param cacheKey 缓存key
     * @param hashKey key-value对
     * @return
     */
    public byte[] getRedisHashCache(final String group, final String cacheKey, final String hashKey){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(cacheKey) && StringUtils.isNotBlank(hashKey)){
            return redisTemplate.execute(new RedisCallback<byte[]>() {
                @Override
                public byte[] doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + cacheKey);
                    byte[] hashKeyBytes = redisTemplate.getStringSerializer().serialize(hashKey);
                    List<byte[]> rtnList = connection.hMGet(keyBytes, hashKeyBytes);
                    if(CollectionUtils.isNotEmpty(rtnList)){
                        return rtnList.get(0);
                    }else{
                        return null;
                    }
                }
            });
        }
        return null;
    }
    
    /**
     * 根据hashKey的集合批量返回缓存值
     * @param group 分组名
     * @param cacheKey 缓存key
     * @param hashKeyList hashKey集合
     * @return
     */
    public List<String> getRedisHashCache(final String group, final String cacheKey, final List<String> hashKeyList){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(cacheKey) && CollectionUtils.isNotEmpty(hashKeyList)){
            return redisTemplate.execute(new RedisCallback<List<String>>() {
                @Override
                public List<String> doInRedis(RedisConnection connection) throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + cacheKey);
                    List<byte[]> hashKeyByteList = new ArrayList<>();
                    for(String hashKey : hashKeyList){
                        hashKeyByteList.add(redisTemplate.getStringSerializer().serialize(hashKey));
                    }
                    byte[][] paramBytes = new byte[hashKeyByteList.size()][];
                    List<byte[]> rtnList = connection.hMGet(keyBytes, hashKeyByteList.toArray(paramBytes));
                    List<String> rtnStrList = new ArrayList<>();
                    if(CollectionUtils.isNotEmpty(rtnList)){
                        for(byte[] bs : rtnList){
                            rtnStrList.add(redisTemplate.getStringSerializer().deserialize(bs));
                        }
                        return rtnStrList;
                    }else{
                        return null;
                    }
                }
            });
        }
        return null;
    }
    
    /**
     * 迭代hash里面的元素
     * @param group
     * @param cacheKey
     * @param count 每次迭代的元素数量
     */
    public List<Map<String, byte[]>> scanRedisHashCache(final String group, final String cacheKey, final long count){
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(cacheKey)){
            return redisTemplate.execute(new RedisCallback<List<Map<String, byte[]>>>() {
                @Override
                public List<Map<String, byte[]>> doInRedis(RedisConnection connection) throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + cacheKey);
                    ScanOptionsBuilder builder = new ScanOptionsBuilder();
                    builder.count(count);
                    ScanOptions options = builder.build();
                    Cursor<Entry<byte[], byte[]>> cursor = connection.hScan(keyBytes, options);
                    List<Map<String, byte[]>> rtnList = new ArrayList<>();
                    while(cursor.hasNext()){
                        Entry<byte[], byte[]> entry = cursor.next();
                        if(entry != null){
                            String keyStr = redisTemplate.getStringSerializer().deserialize(entry.getKey());
                            Map<String, byte[]> map = new HashMap<>();
                            map.put(keyStr, entry.getValue());
                            rtnList.add(map);
                        }
                    }
                    return rtnList;
                }
            });
        }
        return null;
    }
    
    /**
     * 将一个值和其对应的分数放入redis 有序集合缓存中
     * @param group 分组name
     * @param key 分组key
     * @param value 缓存值
     * @param score 分值，可用于排序
     * @return
     */
    public boolean putRedisSortedSet(final String group, final String key, final String value, final double score){
        boolean flag = false;
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            flag = redisTemplate.execute(new RedisCallback<Boolean>() {
                @Override
                public Boolean doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    return connection.zAdd(keyBytes, score, valueBytes);
                }
                
            });
        }
        return flag;
    }
    
    /**
     * 从有序集合中获取对应缓存值的score分值，如果不存在那么返回null
     * @param group 分组name
     * @param key 分组key
     * @param value 缓存值
     * @return
     */
    public Double getRedisSortedSetScore(final String group, final String key, final String value){
        Double score = null;
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            score = redisTemplate.execute(new RedisCallback<Double>() {
                @Override
                public Double doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    return connection.zScore(keyBytes, valueBytes);
                }
                
            });
        }
        return score;
    }
    
    /**
     * 从有序集合中获取对应缓存值的对应的下标，如果不存在那么返回null
     * @param group 分组name
     * @param key 分组key
     * @param value 缓存值
     * @return
     */
    public Long getRedisSortedSetIndex(final String group, final String key, final String value){
        Long index = null;
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)){
            index = redisTemplate.execute(new RedisCallback<Long>() {
                
                @Override
                public Long doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    byte[] valueBytes = redisTemplate.getStringSerializer().serialize(value);
                    return connection.zRank(keyBytes, valueBytes);
                }
            });
        }
        return index;
    }
    
    /**
     * 从有序集合中取出前maxNum条缓存值返回，已经按分数由高到低排好序，maxNum建议不建议1000条
     * @param group 分组name
     * @param key 分组key
     * @param maxNum 最大条数
     * @return
     */
    public List<String> getRedisSortedSetDesc(final String group, final String key, final long maxNum){
        List<String> rtnList = null;
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key) && maxNum > 0 && maxNum <= 1000){
            rtnList = redisTemplate.execute(new RedisCallback<List<String>>() {
                @Override
                public List<String> doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    Set<byte[]> rtnSet = connection.zRevRange(keyBytes, 0, maxNum);
                    if(CollectionUtils.isNotEmpty(rtnSet)){
                        List<String> rtnList = new ArrayList<String>();
                        for(byte[] b : rtnSet){
                            String value = redisTemplate.getStringSerializer().deserialize(b);
                            rtnList.add(value);
                        }
                        return rtnList;
                    }
                    return null;
                }
            });
        }
        return rtnList;
    }
    
    /**
     * 分页查找大于某个分值的缓存值列表
     * @param group 分组名
     * @param key 缓存key
     * @param score 分数
     * @param start 当前页
     * @param pageSize 每页大小
     * @return
     */
    public List<String> getRedisSortedSetByPage(final String group, final String key, final double score, final int start, final int pageSize){
        List<String> rtnList = null;
        if(pageSize > 1000){
            throw new RuntimeException("pageSize can not be more than 1000!");
        }
        if(StringUtils.isNotBlank(group) && StringUtils.isNotBlank(key)){
            rtnList = redisTemplate.execute(new RedisCallback<List<String>>() {
                @Override
                public List<String> doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(siteProperties.getAppName() + GROUP_SEPERATOR + group + GROUP_SEPERATOR + key);
                    Set<byte[]> rtnSet = connection.zRevRangeByScore(keyBytes, score, 99999, start, pageSize);
                    if(CollectionUtils.isNotEmpty(rtnSet)){
                        List<String> rtnList = new ArrayList<String>();
                        for(byte[] b : rtnSet){
                            String value = redisTemplate.getStringSerializer().deserialize(b);
                            rtnList.add(value);
                        }
                        return rtnList;
                    }
                    return null;
                }
            });
        }
        return rtnList;
    }
    
    /**
     * 删除符合pattern规则的所有key
     * @param pattern 可支持正则表达式
     */
    public void deleteRedisCacheByPattern(final String pattern){
        if(StringUtils.isNotBlank(pattern)){
            redisTemplate.execute(new RedisCallback<Boolean>() {
                @Override
                public Boolean doInRedis(RedisConnection connection)
                      throws DataAccessException {
                    byte[] bytes = redisTemplate.getStringSerializer().serialize(pattern);
                    Set<byte[]> keySet = connection.keys(bytes);
                    if(CollectionUtils.isNotEmpty(keySet)){
                        for(byte[] tempBytes : keySet){
                            connection.del(tempBytes);
                        }
                    }
                    return true;
                }
            });
        }
    }
    
    /**
     * 往一个topic里面发送一条消息
     * @param topic topic名称
     * @param message 消息内容
     */
    public void publishTopicMessage(String topic, String message){
        logger.info("publishTopicMessage receive Message, topic:{}, Message:{}, at time:{}", new Object[]{topic, message, DateTimeUtility.timeofnow()});
        if(StringUtils.isNotBlank(topic) && StringUtils.isNotBlank(message)){
            redisTemplate.convertAndSend(topic, message);
        }
    }
    
    public RedisTemplate<String, Object> getRedisTemplate() {
        return redisTemplate;
    }
}
